package org.chen.config;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Component
public class CanalClient implements CommandLineRunner {

    @Value("${canal.host}")
    private String host;

    @Value("${canal.port}")
    private Integer port;

    @Value("${canal.destination}")
    private String destination;

    @Value("${canal.username}")
    private String username;

    @Value("${canal.password}")
    private String password;

    @Autowired
    @Qualifier("slaveTemplate")
    private JdbcTemplate slaveTemplate;

    @Autowired
    @Qualifier("slaveTemplate1")
    private JdbcTemplate slaveTemplate1;

    @Override
    public void run(String... args) throws Exception {
        // 创建canal连接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(host, port),
                destination,
                username,
                password);

        try {
            // 添加连接日志
            log.info("开始连接Canal Server: {}:{}", host, port);
            connector.connect();
            log.info("Canal Server连接成功");

            // 订阅数据库表，格式：database.table
            // 添加订阅日志
            log.info("开始订阅数据库表: flyingorder\\..*");
            connector.subscribe(".*\\..*");
            log.info("订阅成功");
            // 回滚到未进行ack的地方
            connector.rollback();

            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(1000);
                long batchId = message.getId();
                log.info("获取到数据, batchId: {}, size: {}", batchId, message.getEntries().size());
                try {
                    List<CanalEntry.Entry> entries = message.getEntries();
                    if (batchId != -1 && entries.size() > 0) {
                        entries.forEach(this::processEntry);
                    }
                    connector.ack(batchId);
                } catch (Exception e) {
                    log.error("处理数据失败", e);
                    connector.rollback(batchId);
                }
                Thread.sleep(1000);
            }
        } finally {
            connector.disconnect();
        }
    }

    private void processEntry(CanalEntry.Entry entry) {
        if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN ||
                entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
            return;
        }

        String schema = entry.getHeader().getSchemaName();
        String table = entry.getHeader().getTableName();

        CanalEntry.RowChange rowChange;
        try {
            rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        } catch (Exception e) {
            log.error("解析数据失败 ", e);
            return;
        }

        CanalEntry.EventType eventType = rowChange.getEventType();
        JdbcTemplate targetTemplate = getTargetTemplate(schema);
        if (targetTemplate == null) {
            log.warn("Unknown schema: {}", schema);
            return;
        }

        log.info("数据变更: database[{}], table[{}], eventType[{}]", schema, table, eventType);

        rowChange.getRowDatasList().forEach(rowData -> {
            switch (eventType) {
                case INSERT:
                    syncInsert(targetTemplate, table, rowData.getAfterColumnsList());
                    break;
                case UPDATE:
                    syncUpdate(targetTemplate, table, rowData.getBeforeColumnsList(), rowData.getAfterColumnsList());
                    break;
                case DELETE:
                    syncDelete(targetTemplate, table, rowData.getBeforeColumnsList());
                    break;
                default:
                    break;
            }
        });
    }

    private JdbcTemplate getTargetTemplate(String schema) {
        switch (schema) {
            case "flyingorder":
                return slaveTemplate;
            case "flyingorder_1":
                return slaveTemplate1;
            default:
                return null;
        }
    }

    private void syncInsert(JdbcTemplate template, String table, List<CanalEntry.Column> columns) {
        List<String> columnNames = new ArrayList<>();
        List<String> columnValues = new ArrayList<>();

        for (CanalEntry.Column column : columns) {
            // 处理MySQL关键字
            String columnName = escapeKeyword(column.getName());
            columnNames.add(columnName);
            String value = formatValue(column.getValue(), column.getMysqlType());
            columnValues.add(value);
        }

        String sql = String.format("INSERT INTO %s (%s) VALUES (%s)",
                table,
                String.join(", ", columnNames),
                String.join(", ", columnValues));

        executeWithRetry(template, sql);
    }

    private void syncUpdate(JdbcTemplate template, String table,
                            List<CanalEntry.Column> beforeColumns,
                            List<CanalEntry.Column> afterColumns) {
        List<String> setClauses = new ArrayList<>();
        String whereClause = null;

        for (CanalEntry.Column column : afterColumns) {
            if (column.getUpdated()) {
                String columnName = escapeKeyword(column.getName());
                String value = formatValue(column.getValue(), column.getMysqlType());
                setClauses.add(String.format("%s = %s",
                        columnName, value));
            }
        }

        for (CanalEntry.Column column : beforeColumns) {
            if (column.getIsKey()) {
                String columnName = escapeKeyword(column.getName());
                String value = formatValue(column.getValue(), column.getMysqlType());
                whereClause = String.format("%s = %s",
                        columnName, value);
                break;
            }
        }

        if (whereClause != null) {
            String sql = String.format("UPDATE %s SET %s WHERE %s",
                    table,
                    String.join(", ", setClauses),
                    whereClause);

            executeWithRetry(template, sql);
        }
    }

    private void syncDelete(JdbcTemplate template, String table, List<CanalEntry.Column> columns) {
        String whereClause = columns.stream()
                .filter(CanalEntry.Column::getIsKey)
                .map(col -> {
                    String columnName = escapeKeyword(col.getName());
                    String value = formatValue(col.getValue(), col.getMysqlType());
                    return String.format("%s = %s", columnName, value);
                })
                .collect(Collectors.joining(" AND "));

        String sql = String.format("DELETE FROM %s WHERE %s", table, whereClause);

        executeWithRetry(template, sql);
    }

    /**
     * 格式化SQL值，处理不同的数据类型
     */
    private String formatValue(String value, String mysqlType) {
        if (value == null || value.isEmpty()) {
            return "NULL";
        }

        // 处理不同的MySQL数据类型
        mysqlType = mysqlType.toUpperCase();
        if (mysqlType.contains("DATETIME") || mysqlType.contains("TIMESTAMP")) {
            return value.isEmpty() ? "NULL" : String.format("'%s'", escapeString(value));
        } else if (mysqlType.contains("DATE")) {
            return value.isEmpty() ? "NULL" : String.format("'%s'", escapeString(value));
        } else if (mysqlType.contains("INT") || mysqlType.contains("DECIMAL") ||
                mysqlType.contains("FLOAT") || mysqlType.contains("DOUBLE")) {
            return value.isEmpty() ? "NULL" : value;
        } else {
            // 字符串类型
            return String.format("'%s'", escapeString(value));
        }
    }

    /**
     * SQL注入防护
     */
    private String escapeString(String value) {
        if (value == null) {
            return null;
        }
        return value.replace("'", "''");
    }

    /**
     * 处理MySQL关键字
     */
    private String escapeKeyword(String columnName) {
        // MySQL关键字列表
        Set<String> keywords = new HashSet<>(Arrays.asList(
                "unique", "order", "desc", "group", "key", "like", "by",
                "index", "table", "database", "select", "update", "insert", "delete"
        ));

        if (keywords.contains(columnName.toLowerCase())) {
            return "`" + columnName + "`";
        }
        return columnName;
    }

    /**
     * 添加日志记录
     */
    private void executeWithRetry(JdbcTemplate template, String sql) {
        int maxRetries = 3;
        int retryCount = 0;
        while (retryCount < maxRetries) {
            try {
                log.info("Executing SQL: {}", sql);
                template.execute(sql);
                log.info("SQL executed successfully");
                return;
            } catch (Exception e) {
                retryCount++;
                log.error("Error executing SQL (attempt {}/{}): {}\nSQL: {}",
                        retryCount, maxRetries, e.getMessage(), sql);
                if (retryCount == maxRetries) {
                    throw new RuntimeException("Failed to execute SQL after " + maxRetries +
                            " attempts. SQL: " + sql, e);
                }
                try {
                    Thread.sleep(1000 * retryCount);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(ie);
                }
            }
        }
    }
}
